package com.abel.bigwater.wflow

import com.abel.bigwater.wflow.mapper.UserMapper
import com.alibaba.fastjson.JSON
import org.activiti.api.process.runtime.ProcessRuntime
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import org.junit.FixMethodOrder
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.test.context.junit4.SpringRunner
import java.util.*
import java.util.concurrent.TimeUnit

@RunWith(SpringRunner::class)
@SpringBootTest
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class BwWflowApplicationTests {
    @Autowired
    var userMapper: UserMapper? = null

    @Autowired
    var processRuntime: ProcessRuntime? = null

    // @Autowired
    var mqTemp: KafkaTemplate<Any, Any>? = null

    // @Autowired
    var consumerFactory: ConsumerFactory<Any, Any>? = null

    @Test
    fun contextLoads() {
    }

    @Test
    fun test12User() {
        lgr.info("processRuntime: ${processRuntime?.configuration()}")
        val ulist = userMapper?.listUser("1", null, null)
        lgr.info("users : ${JSON.toJSONString(ulist, true)} from mapper: $userMapper")
    }

    @Test
    fun test21MsgQueue() {
        var consumer = consumerFactory!!.createConsumer("app", "")
        consumer.subscribe(listOf("app_log"), object : ConsumerRebalanceListener {
            /**
             * A callback method the user can implement to provide handling of customized offsets on completion of a successful
             * partition re-assignment. This method will be called after an offset re-assignment completes and before the
             * consumer starts fetching data.
             *
             *
             * It is guaranteed that all the processes in a consumer group will execute their
             * [.onPartitionsRevoked] callback before any instance executes its
             * [.onPartitionsAssigned] callback.
             *
             *
             * It is common for the assignment callback to use the consumer instance in order to query offsets. It is possible
             * for a [org.apache.kafka.common.errors.WakeupException] or [org.apache.kafka.common.errors.InterruptException]
             * to be raised from one these nested invocations. In this case, the exception will be propagated to the current
             * invocation of [KafkaConsumer.poll] in which this callback is being executed. This means it is not
             * necessary to catch these exceptions and re-attempt to wakeup or interrupt the consumer thread.
             *
             * @param partitions The list of partitions that are now assigned to the consumer (may include partitions previously
             * assigned to the consumer)
             * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to [KafkaConsumer]
             * @throws org.apache.kafka.common.errors.InterruptException If raised from a nested call to [KafkaConsumer]
             */
            override fun onPartitionsAssigned(partitions: MutableCollection<TopicPartition>?) {
                lgr.info("parts: ${partitions?.joinToString()}")
            }

            /**
             * A callback method the user can implement to provide handling of offset commits to a customized store on the start
             * of a rebalance operation. This method will be called before a rebalance operation starts and after the consumer
             * stops fetching data. It is recommended that offsets should be committed in this callback to either Kafka or a
             * custom offset store to prevent duplicate data.
             *
             *
             * For examples on usage of this API, see Usage Examples section of [KafkaConsumer]
             *
             *
             * **NOTE:** This method is only called before rebalances. It is not called prior to [KafkaConsumer.close].
             *
             *
             * It is common for the revocation callback to use the consumer instance in order to commit offsets. It is possible
             * for a [org.apache.kafka.common.errors.WakeupException] or [org.apache.kafka.common.errors.InterruptException]
             * to be raised from one these nested invocations. In this case, the exception will be propagated to the current
             * invocation of [KafkaConsumer.poll] in which this callback is being executed. This means it is not
             * necessary to catch these exceptions and re-attempt to wakeup or interrupt the consumer thread.
             *
             * @param partitions The list of partitions that were assigned to the consumer on the last rebalance
             * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to [KafkaConsumer]
             * @throws org.apache.kafka.common.errors.InterruptException If raised from a nested call to [KafkaConsumer]
             */
            override fun onPartitionsRevoked(partitions: MutableCollection<TopicPartition>?) {
                lgr.info("revoked: ${partitions?.joinToString()}")
            }

        })

        val ft = mqTemp!!.send("app_log", "Test:" + UUID.randomUUID().toString())
        val sr = ft[2, TimeUnit.SECONDS]
        ft.addCallback({ sr ->
            lgr.info("send result: ${sr}")
        }, { ex ->
            lgr.error("Fail to send caused by ${ex.message}", ex)
        })

        Thread.sleep(2 * 1000)
    }

    @Test
    fun test22Consumer() {
        var consumer = consumerFactory!!.createConsumer("app", "")
        val tlist = consumer.listTopics()

        lgr.info("topics: ${tlist.values.joinToString()}")

        consumer.subscribe(listOf("app_log"))
        consumer.poll(10).forEach {
            lgr.info("consumer got: ${it}")
        }
    }

    companion object {
        private val lgr = LoggerFactory.getLogger(BwWflowApplicationTests::class.java)
    }

}
